package com.hub.realtime.sqlclient.base;

import com.alibaba.fastjson.JSON;
import com.hub.realtime.sqlclient.base.model.FlinkCatalog;
import com.hub.realtime.sqlclient.base.model.FlinkRunEnv;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

@Slf4j
public class TableEnvInital {

    public static  FlinkRunEnv createEnv(String[] args){
        FlinkRunEnv env = new FlinkRunEnv();
        ParameterTool parameter = ParameterTool.fromArgs(args);
        env.setParameterTool(parameter);
        EnvironmentSettings.Builder builder = EnvironmentSettings.newInstance();
        if (parameter.get("exeModel").equals("streaming")) {
            builder.inStreamingMode();
        } else {
            builder.inBatchMode();
        }
//        String catalogName = "";
//        String databaseName = "";
//        String catalog = parameter.get("catalog");
//        if (StringUtils.isNotEmpty(catalog)) {
//            FlinkCatalog flinkCatalog = JSON.parseObject(catalog, FlinkCatalog.class);
//            catalogName = flinkCatalog.getCatalogName();
//            databaseName = flinkCatalog.getDatabaseName();
//        }
//        if (StringUtils.isNotEmpty(catalogName)
//                && StringUtils.isNotEmpty(databaseName)) {
//            log.info("with built in catalog: ".concat(catalogName));
//            log.info("with built in database: ".concat(databaseName));
//            builder.withBuiltInCatalogName(catalogName);
//            builder.withBuiltInDatabaseName(databaseName);
//        } else if (StringUtils.isNotEmpty(catalogName) && StringUtils.isEmpty(databaseName)) {
//            log.info("with built in catalog: ".concat(catalogName));
//            builder.withBuiltInCatalogName(catalogName);
//        } else if (StringUtils.isEmpty(catalogName) && StringUtils.isNotEmpty(databaseName)) {
//            log.info("with built in database: ".concat(databaseName));
//            builder.withBuiltInDatabaseName(databaseName);
//        }

        try {
            Class<?> aClass = Class.forName("org.apache.flink.table.planner.delegation.PlannerContext");
            log.info("发现类："+aClass);
        }catch (Exception ex)
        {
            log.error(ex.getMessage());
        }


        EnvironmentSettings environmentSettings = builder.build();
        if (parameter.get("exeModel").equals("streaming")) {
            StreamExecutionEnvironment senv = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setTableEnvironment(StreamTableEnvironment.create(senv, environmentSettings));
            env.setSenv(senv);
        } else {
            env.setTableEnvironment(TableEnvironment.create(environmentSettings));
        }
        return env;
    }


    public static void main(String[] args) {


        try {
            Class<?> aClass = Class.forName("org.apache.flink.table.planner.delegation.PlannerContext");
            log.info("发现类："+aClass);
        }catch (Exception ex)
        {
            log.error(ex.getMessage());
        }

    }


}
